package com.g.ocache.impl.store.expiration;

import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.apache.log4j.Logger;

import com.g.ocache.utils.XThreadFactory;


/**
 * 
 * @author Jingqi Xu
 */
public class SmartExpirationEvictor<K, V> implements ExpirationEvictor<K, V> {
	//
	private static final Logger LOGGER = Logger.getLogger(SmartExpirationEvictor.class);
	
	//
	private final int interval;
	private final int threshold;
	private final int adjustment;
	private final Expirable<K, V> expirable;
	private final AtomicInteger currentThreshold;
	private final ScheduledExecutorService executor;
	private final AtomicBoolean verbose = new AtomicBoolean(true);
	private final AtomicBoolean started = new AtomicBoolean(false);
	
	/**
	 * 
	 */
	public SmartExpirationEvictor(Expirable<K, V> expirable, int interval, int threshold) {
		this(expirable, interval, threshold, 0);
	}
	
	public SmartExpirationEvictor(Expirable<K, V> expirable, int interval, int threshold, int adjustment) {
		this(expirable, interval, threshold, adjustment, new XThreadFactory(SmartExpirationEvictor.class.getSimpleName(), true));
	}
	
	public SmartExpirationEvictor(Expirable<K, V> expirable, int interval, int threshold, int adjustment, ThreadFactory tf) {
		this(expirable, interval, threshold, adjustment, Executors.newScheduledThreadPool(1, tf));
	}
	
	public SmartExpirationEvictor(Expirable<K, V> expirable, int interval, int threshold, int adjustment, ScheduledExecutorService executor) {
		//
		this.expirable = expirable;
		this.interval = interval;
		this.threshold = threshold;
		this.adjustment = (adjustment == 0 ? (int)(threshold * 0.15) : adjustment);
		this.executor = executor;
		
		//
		this.currentThreshold = new AtomicInteger(threshold);
	}
	
	/**
	 * 
	 */
	public String toString() {
		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
		.append("interval", interval)
		.append("verbose", isVerbose())
		.append("running", isRunning())
		.append("expirable", expirable)
		.append("threshold", threshold)
		.append("adjustment", adjustment)
		.append("currentThreshold", currentThreshold.get()).toString();
	}
	
	/**
	 * 
	 */
	public boolean isRunning() {
		return started.get();
	}
	
	public void start() {
		// Precondition checking
		if(!started.compareAndSet(false, true)) {
			throw new IllegalStateException(this + " has already been started");
		}
		
		//
		this.executor.scheduleWithFixedDelay(new Task(), interval, interval, TimeUnit.SECONDS);
	}

	public void stop() {
		// Precondition checking
		if(!started.compareAndSet(true, false)) {
			return;
		}
		
		//
		executor.shutdownNow();
	}
	
	/**
	 * 
	 */
	public int getInterval() {
		return interval;
	}
	
	public int getThreshold() {
		return threshold;
	}
	
	public int getAdjustment() {
		return adjustment;
	}
	
	public int getCurrentThreshold() {
		return currentThreshold.get();
	}
	
	public boolean isVerbose() {
		return verbose.get();
	}

	public void setVerbose(boolean verbose) {
		this.verbose.set(verbose);
	}
	
	public Expirable<K, V> getExpirable() {
		return expirable;
	}
	
	/**
	 * 
	 */
	public static class FactoryImpl<K, V> implements Factory<K, V> {
		//
		private int interval;
		private int threshold;
		private int adjustment;

		/**
		 * 
		 */
		public FactoryImpl(int interval, int threshold) {
			this(interval, threshold, 0);
		}
		
		public FactoryImpl(int interval, int threshold, int adjustment) {
			this.interval = interval;
			this.threshold = threshold;
			this.adjustment = adjustment;
		}

		/**
		 * 
		 */
		public ExpirationEvictor<K, V> create(Expirable<K, V> expirable) {
			return new SmartExpirationEvictor<K, V>(expirable, interval, threshold, adjustment);
		}

		/**
		 * 
		 */
		public int getInterval() {
			return interval;
		}

		public void setInterval(int interval) {
			this.interval = interval;
		}
		
		public int getThreshold() {
			return threshold;
		}

		public void setThreshold(int threshold) {
			this.threshold = threshold;
		}
		
		public int getAdjustment() {
			return adjustment;
		}

		public void setAdjustment(int adjustment) {
			this.adjustment = adjustment;
		}
	}
	
	private class Task implements Runnable {
		
		public void run() {
			try {
				//
				if(expirable == null) {
					LOGGER.warn("can not schedule " + this + " because expirable is null");
					return;
				}
				
				// Evict
				if(expirable.size() >= getCurrentThreshold()) {
					//
					long now = System.currentTimeMillis();
					Map<K, V> expired = expirable.detect();
					
					//
					if(isVerbose() && LOGGER.isInfoEnabled()) {
						LOGGER.info(expirable + " was successfully evicted, count: " + expired.size() + ", time elapsed: " + (System.currentTimeMillis() - now));
					}
				}
				
				// Adjust threshold
				boolean adjusted = false;
				final int size = expirable.size();
				final int threshold = getCurrentThreshold();
				if(size > threshold) { // Need to increment
					adjusted = currentThreshold.compareAndSet(threshold, threshold + getAdjustment()); 
				} else if(size < (threshold - getAdjustment())) { // Need to decrement
					if(threshold - getAdjustment() > getThreshold()) {
						adjusted = currentThreshold.compareAndSet(threshold, threshold - getAdjustment());
					}
				}
				
				//
				if(adjusted) {
					if(isVerbose() && LOGGER.isInfoEnabled()) {
						LOGGER.info("threshold was adjusted, before: " + threshold + ", after: " + getCurrentThreshold() + ", expirable: " + expirable);
					}
				}
			} catch(Exception e) {
				LOGGER.error("unhandled exception in expiration evictor: " + SmartExpirationEvictor.this, e);
			}
		}
	}
}
